package xin.alum.aim.cluster;

import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;
import xin.alum.aim.constant.Constants;
import xin.alum.aim.groups.ClusterFactory;
import xin.alum.aim.groups.Session;
import xin.alum.aim.model.Pusher;
import xin.alum.aim.model.Transportable;
import java.util.Comparator;
import java.util.concurrent.PriorityBlockingQueue;

public abstract class ClusterPusher implements ClusterFactory {

    protected final InternalLogger logger = InternalLoggerFactory.getInstance(this.getClass());

    protected abstract <T> boolean push(String channel, T d);

    private Comparator<Pusher<Transportable>> comparator = (p1, p2) -> p1.getData().getPriority() > p2.getData().getPriority() ? p1.getData().getPriority() : p2.getData().getPriority();

    protected PriorityBlockingQueue<Pusher<Transportable>> queue = new PriorityBlockingQueue(ClusterFactory.QUEUE_CAPACITY, comparator);


    protected abstract  void pushQueue();
    @Override
    public void kick(Session session) {
        push(Constants.BIND_MESSAGE_INNER_QUEUE, session);
    }

    @Override
    public void push(String recipient, Transportable data) {
        Pusher p = new Pusher(recipient, data);
        queue.put(p);
        pushQueue();
    }
}
